package cn.csc.module.pay.service.notify;

import cn.csc.framework.common.constants.GlobalConstant;
import cn.csc.module.pay.dal.dataobject.notify.NotifyRefundLogDO;
import cn.csc.module.pay.dal.dataobject.notify.PayNotifyTaskDO;
import cn.csc.module.pay.dal.mysql.notify.NotifyRefundLogMapper;
import cn.csc.module.pay.dal.redis.notify.PayNotifyLockRedisDAO;
import cn.csc.module.pay.enums.notify.PayNotifyStatusEnum;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.http.HttpUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

import org.springframework.validation.annotation.Validated;

import java.io.Serializable;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;


import static cn.csc.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.csc.module.pay.framework.job.config.PayJobConfiguration.THREAD_POOL_TASK_EXECUTOR;

/**
 * 支付回调本地通知 App 的日志 Service 实现类
 *
 * @author 曾铁树
 */
@Slf4j
@Service
@Validated
public class NotifyRefundLogServiceImpl implements NotifyRefundLogService {

    @Resource
    private NotifyRefundLogMapper notifyRefundLogMapper;
    @Resource(name = THREAD_POOL_TASK_EXECUTOR)
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Resource
    private PayNotifyLockRedisDAO payNotifyLockCoreRedisDAO;


    @Override
    public int executeNotify() throws InterruptedException {
        List<NotifyRefundLogDO> tasks = getNotifyRefundLogList();
        if (CollUtil.isEmpty(tasks)) {
            return 0;
        }
        // 遍历，逐个通知
        CountDownLatch latch = new CountDownLatch(tasks.size());
        tasks.forEach(task -> threadPoolTaskExecutor.execute(() -> {
            try {
                executeNotifySync(task);
            } finally {
                latch.countDown();
            }
        }));
        // 等待完成
        awaitExecuteNotify(latch);
        // 返回执行完成的任务数（成功 + 失败)
        return tasks.size();

    }

    @Override
    public NotifyRefundLogDO selectById(Serializable id) {
        return notifyRefundLogMapper.selectById(id);
    }

    @Override
    public int insert(NotifyRefundLogDO notifyRefundLogDO) {
        return notifyRefundLogMapper.insert(notifyRefundLogDO);
    }

    @Override
    public int updateById(NotifyRefundLogDO notifyRefundLogDO) {
        return notifyRefundLogMapper.updateById(notifyRefundLogDO);
    }

    @Override
    public List<NotifyRefundLogDO> getNotifyRefundLogList() {
        return notifyRefundLogMapper.selectListByNotifyRefund();
    }

    /**
     * 等待全部支付通知的完成
     * 每 1 秒会打印一次剩余任务数量
     *
     * @param latch Latch
     * @throws InterruptedException 如果被打断
     */
    private void awaitExecuteNotify(CountDownLatch latch) throws InterruptedException {
        long size = latch.getCount();
        for (int i = 0; i < GlobalConstant.TIMEOUT; i++) {
            if (latch.await(1L, TimeUnit.SECONDS)) {
                return;
            }
            log.info("[awaitExecuteNotify][任务处理中， 总任务数({}) 剩余任务数({})]", size, latch.getCount());
        }
        log.error("[awaitExecuteNotify][任务未处理完，总任务数({}) 剩余任务数({})]", size, latch.getCount());
    }

    /**
     * 同步执行单个支付通知
     *
     * @param task 通知任务
     */
    public void executeNotifySync(NotifyRefundLogDO task) {
        // 分布式锁，避免并发问题
        payNotifyLockCoreRedisDAO.lock("notify:refund:ben:" + task.getId(), GlobalConstant.TIMEOUT, () -> {
            // 校验，当前任务是否已经被通知过
            // 虽然已经通过分布式加锁，但是可能同时满足通知的条件，然后都去获得锁。此时，第一个执行完后，第二个还是能拿到锁，然后会再执行一次。
            String resultStr = HttpUtil.createPost(task.getNotifyUrl()).body(task.getResponse()).timeout((int) GlobalConstant.TIMEOUT).execute().body();

            if (!"success".equals(resultStr)) {
                //通过插入数据库记录通知出错的记录
                if (Objects.nonNull(task)) {
                    NotifyRefundLogDO logDO = NotifyRefundLogDO.builder()
                            .id(task.getId()).build();
                    if (logDO.getNotifyTimes() >= PayNotifyTaskDO.NOTIFY_FREQUENCY.length) {
                        logDO.setStatus(PayNotifyStatusEnum.FAILURE.getStatus());
                    } else {
                        logDO.setNotifyTimes(task.getNotifyTimes() + 1);
                    }
                    notifyRefundLogMapper.updateById(logDO);
                }
            } else {
                //通知成功处理
                NotifyRefundLogDO logDO = NotifyRefundLogDO.builder().status(PayNotifyStatusEnum.SUCCESS.getStatus())
                        .id(task.getId()).notifyTimes(task.getNotifyTimes() + 1)
                        .status(PayNotifyStatusEnum.SUCCESS.getStatus()).build();
                notifyRefundLogMapper.updateById(logDO);
            }

        });
    }

}
